package yatan.cluster;

import java.util.List;

import java.util.Properties;

import org.jgroups.Address;
import org.jgroups.Message;
import org.jgroups.SuspectedException;
import org.jgroups.TimeoutException;

import yatan.cluster.concurrent.JobDoneHandler;
import yatan.cluster.concurrent.JobPool;
import yatan.cluster.job.Job;
import yatan.cluster.job.JobResult;
import yatan.cluster.job.JobRunnerFactory;
import yatan.common.utility.ParamUtility;
import yatan.common.ConfigUtility;

public class ServantNode extends BaseNode {
    public static final String DEFAULT_CONFIG_FILE = "servant_node.properties";

    private int threadPoolSize = 200;

    private int requestOnceJobCount = 200;

    private int findMasterNodeDelay = 3 * 1000;
    private int waitJobScheduleDelay = 60 * 1000;
    private int jobHandleTimeout = 30 * 1000;

    private JobPool jobPool;

    private Address masterNodeAddress;

    private Boolean dispathJobRequestSent = Boolean.FALSE;

    public ServantNode() {
        // config the node with default property file
        this(DEFAULT_CONFIG_FILE);
    }

    public ServantNode(String configFile) {
        ParamUtility.checkStringNullEmpty(configFile, "configFile");

        // load configuration file
        Properties properties = ConfigUtility.loadConfigFile(configFile);

        // config the node with the loaded property file
        config(properties);

    }

    public ServantNode(Properties properties) {
        ParamUtility.checkNull(properties, "properties");

        // config the node with given properties
        config(properties);
    }

    @SuppressWarnings("unchecked")
    @Override
    protected void work() throws NodeException {
        // find the master node
        findMasterNode();

        while (isStarted()) {
            if (this.jobPool.getJobCount() < this.threadPoolSize * 1.3) {
                getLogger().info(
                        "Running jobs " + this.jobPool.getJobCount() + " < " + this.threadPoolSize * 1.3
                                + " on the node, requesting new jobs...");

                NodeMessage nodeMessage = new NodeMessage();
                nodeMessage.setNodeMessageType(NodeMessageType.GET_JOBS);
                nodeMessage.setData(this.requestOnceJobCount);

                List<Job> retrievedJobs = null;
                do
                    try {
                        retrievedJobs =
                                (List<Job>) sendMessage(this.masterNodeAddress, nodeMessage, List.class,
                                        this.waitJobScheduleDelay);
                    } catch (TimeoutException e) {
                        getLogger().warn("Retrieving jobs timeout. Try again...");
                    } catch (SuspectedException e) {
                        getLogger().warn("Master is suspected off line.");
                        findMasterNode();
                    }
                while (retrievedJobs == null);

                getLogger().debug("Received " + retrievedJobs.size() + " jobs. Submiting to thread pool...");
                for (Job job : retrievedJobs) {
                    this.jobPool.submit(job);
                }
                getLogger().info("Receieved " + retrievedJobs.size() + " jobs.");
            }

            getLogger().debug(
                    "Wait until another job is done(" + (this.waitJobScheduleDelay / 1000) + " seconds most)...");
            synchronized (this) {
                try {
                    wait(this.waitJobScheduleDelay);
                } catch (InterruptedException e) {
                    getLogger().warn("Cannot wait on the node object. Interrupted exception: " + e.getMessage(), e);
                }
            }
        }
    }

    private void config(Properties properties) {
        // read cluster name
        setClusterName(ConfigUtility.readStringProperty(properties, "cluster_name", false, DEFAULT_CLUSTER_NAME));
        // read JGroup config path
        setJgroupConfig(ConfigUtility.readStringProperty(properties, "jgroup_config", false, null));

        // read config properties
        this.threadPoolSize = ConfigUtility.readIntProperty(properties, "thread_pool_size", false, 100);
        this.requestOnceJobCount = ConfigUtility.readIntProperty(properties, "request_once_job_count", false, 100);
        this.findMasterNodeDelay = ConfigUtility.readIntProperty(properties, "find_master_node_delay", false, 5000);
        this.waitJobScheduleDelay = ConfigUtility.readIntProperty(properties, "wait_job_schedule_delay", false, 60000);
        this.jobHandleTimeout = ConfigUtility.readIntProperty(properties, "job_handle_timeout", false, this.jobHandleTimeout);

        // create job runner factory
        JobRunnerFactory jobRunnerFactory =
                ConfigUtility.createObject(properties, "job_runner_factory", JobRunnerFactory.class);

        // initialize job pool
        this.jobPool = new JobPool(this.threadPoolSize);
        this.jobPool.setJobRunnerFactory(jobRunnerFactory);
        this.jobPool.setJobDoneHandler(new ServantJobDoneHandler());
    }

    private void findMasterNode() throws NodeException {
        getLogger().info("Try to find master node...");
        do {
            NodeMessage nodeMessage = new NodeMessage();
            nodeMessage.setNodeMessageType(NodeMessageType.FIND_MASTER_BROADCAST);

            getLogger().debug("Sending request to find master node...");
            getLogger().debug("Wait " + this.findMasterNodeDelay + " miliseconds for master node response...");
            try {
                this.masterNodeAddress = sendMessage(null, nodeMessage, Address.class, this.findMasterNodeDelay);
            } catch (TimeoutException e) {
                getLogger().debug("Timeout, retry...");
            } catch (SuspectedException e) {
                getLogger().debug("Timeout, retry...");
            }
        } while (this.masterNodeAddress == null);
        getLogger().info("Found master node succesfully. The address is " + this.masterNodeAddress);
    }

    private class ServantJobDoneHandler implements JobDoneHandler {
        @Override
        public void success(JobResult jobResult) {
            getLogger().debug("Job " + jobResult.getId() + " is done. Sending job result to master node...");
            NodeMessage nodeMessage = new NodeMessage();
            nodeMessage.setNodeMessageType(NodeMessageType.JOB_DONE);
            nodeMessage.setData(jobResult);

            try {
                sendMessage(ServantNode.this.masterNodeAddress, nodeMessage, Object.class, jobHandleTimeout);
            } catch (TimeoutException e) {
                getLogger().warn("Sending JobResult " + jobResult + " to master timeout. Will be discarded.");
            } catch (SuspectedException e) {
                getLogger().warn(
                        "Sending JobResult " + jobResult
                                + " to master timeout. Master is suspected down. Will be discarded.");
            } catch (NodeException e) {
                // never occur
            }

            synchronized (ServantNode.this) {
                ServantNode.this.notify();
            }
        }

        @Override
        public void error(Job job, Exception error) {
            getLogger().debug("An error occured in the job " + job + ". Sending error message to master node...");
            NodeMessage nodeMessage = new NodeMessage();
            nodeMessage.setNodeMessageType(NodeMessageType.JOB_DONE_WITH_ERROR);
            nodeMessage.setData(new Object[] {job, error});

            try {
                sendMessage(ServantNode.this.masterNodeAddress, nodeMessage, Object.class, jobHandleTimeout);
            } catch (TimeoutException e) {
                getLogger().warn("Sending failure message of Job " + job + " to master timeout. Will be discarded.");
            } catch (SuspectedException e) {
                getLogger().warn(
                        "Sending failure message of Job " + job
                                + " to master timeout. Master is suspected down. Will be discarded.");
            } catch (NodeException e) {
                // never occur
            }

            synchronized (ServantNode.this) {
                ServantNode.this.notify();
            }
        }
    }

    @Override
    public Object handle(Message message) {
        // do not process any message
        getLogger().debug("Receieve message :" + message + ", will be disgarded.");
        return null;
    }
}
